package com.mango.ch05;

import java.io.IOException;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class RelativeFrequencyReducer extends Reducer<PairOfWords, IntWritable, PairOfWords, DoubleWritable> {
	private double totalCount = 0;
	private String currentWord = "NO_DEFINED";
	private int times = 0;

	/**
	 * w1 只是wi的另一种形式，表示随机的某个单词
	 */
	@Override
	protected void reduce(PairOfWords key, Iterable<IntWritable> values, Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub

		System.out.println("reduce 当前key是：" + key.getLeftEle() + " 值是：" + key.getSecond());
		System.out.println("-------------- " + times++ + " ----------");
		// ‘*’的值是42 小于字母A=65，所以有((w,*),1) 排序在所有((w,wi),i)之前
		if (key.getRightEle().equals("*")) {
			// map传出的数据格式相当于 ((w1，*),1),而分区器已经按照组合键的左边个值进行分区，所以只要是((w1，-），-）形式的
			// 都会由同一个ruducer处理
			// 所以收到数据格式相当于((w1,*),2),((w1,w2),1),((w1,w3),1),但是rudce一次只会处理同一key的键值对
			// 所以（（w1，-），-）形式的会由同一个ruduce处理,会调用多次reduce

			// 下面的if-else 是判断是否是* 记录w1 在全文的组合词对中出现的总次数
			if (key.getLeftEle().equals(currentWord)) {
				totalCount += getTotalCount(values);
			} else {
				currentWord = key.getLeftEle();// 设置当前处理的分区器键（wi,i)
				totalCount = getTotalCount(values);// 记录总的出现次数
			}
			System.out.println("---if----当前单词" + currentWord + "出现总数" + totalCount);
		} else {
			// 此处的代码是用来表示某单词 与当前word共同出现的概率
			double count = getTotalCount(values);
			double relativeCount = count / totalCount;
			context.write(key, new DoubleWritable(relativeCount));
			System.out.println("---else----当前单词" + currentWord + "出现总数" + totalCount);
		}
		System.out.println("当前total  " + totalCount);
	}

	/**
	 * 获取values 集合中的总值 换句话说就是统计w1出现的总次数
	 * 
	 * @param values
	 * @return
	 */
	private double getTotalCount(Iterable<IntWritable> values) {
		// TODO Auto-generated method stub
		double res = 0;
		System.out.print("值集是：  ");
		for (IntWritable value : values) {
			String intstr = value.toString();
			res += Double.valueOf(intstr);
			System.out.print(value + "  " + " string 后的值是" + value.get());
		}
		return res;
	}

}
